package com.dys.flink.wc1.netty.handler;

import com.dys.flink.wc1.model.Product;
import com.dys.flink.wc1.util.JsonUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

/**
 * 只写 JSON 字节，长度前缀由 LengthFieldPrepender 负责
 */
public class ProductEncoder extends MessageToByteEncoder<Product> {

    private static final Logger logger = LoggerFactory.getLogger(ProductEncoder.class);

    @Override
    protected void encode(ChannelHandlerContext ctx, Product product, io.netty.buffer.ByteBuf out) throws Exception {
        try {
            String json = JsonUtils.toJson(product);
            byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
            out.writeBytes(bytes);      // 不再写长度！
            logger.debug("Successfully encoded product: {}", product);
        } catch (Exception e) {
            logger.error("Failed to encode product: {}", product, e);
            throw e;
        }
    }
}